package com.gugee.tiktok.data.consumer.blogger.listener;

import com.gugee.tiktok.data.common.msg.AwemeMessage;

import com.gugee.tiktok.data.common.msg.BloggerMessage;
import com.gugee.tiktok.data.consumer.blogger.config.DataConsumerConfig;
import com.gugee.tiktok.data.consumer.blogger.core.service.AwemeService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.stream.Collectors;


/**
 *
 * @author huang xin
 * Date  2020/10/04
 */
@Component
@Slf4j
public class AwemeExistMessageListener {

    @Autowired
    AwemeService awemeService;

    @Autowired
    DataConsumerConfig dataConsumerConfig;

    @KafkaListener(id="${kafka.consumer.group.awemeExist.id}",topics = {"${kafka.topics.awemeExist.online}"},containerFactory = "factoryAwemeExistAck")
    public void listen(AwemeMessage messages, Acknowledgment ack) {
        //任务开始时间
        long start = System.currentTimeMillis();
        //任务结果消息
        String error = "success";
        //时间时限
        long acceptMsgTimeSecRange = dataConsumerConfig.getAcceptAwemeExistMsgTimeMinRange() * 60 * 1000;

        try {
            long isRange = start - acceptMsgTimeSecRange;
            //处理
            if(messages.getEventTime() >= isRange){
                awemeService.awemeExistHandle(messages);
            }
            ack.acknowledge();
        }catch (Exception e){
            log.error("CheckAwemeExist 消息处理任务发生异常,awemeId : {} , 异常 ：{}" ,messages.getAwemeId(),e.getMessage());
            error = e.getMessage();
        }finally {
            long times = (System.currentTimeMillis() - start) / 1000;
            if(times < 1){
                times = 1;
            }
            log.info("CheckAwemeExist 消息处理，awemeId: {},处理耗时: {}s,结果：{}" ,
                    messages.getAwemeId(),times,error);
        }
    }
}
